package com.tongtech.proxy.core.protocol.redis.callbacks;

import com.tongtech.proxy.core.protocol.redis.ProcessorRedisImp;
import com.tongtech.proxy.core.utils.ProcessCounter;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import com.tongtech.proxy.core.acl.AclAuthen;
import com.tongtech.proxy.core.protocol.DataResult;
import com.tongtech.proxy.core.protocol.SessionAttribute;
import com.tongtech.proxy.core.protocol.SlowLogs;
import com.tongtech.proxy.core.utils.Log;
import com.tongtech.proxy.core.utils.ProxyConfig;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

import static com.tongtech.proxy.core.StaticContent.CachedSessionAttributes;

public class DefaultDataResultImp implements DataResult {

    private static final Log logger = ProxyConfig.getServerLog();

    private static final HashSet<String> UnCallbackCommand = new HashSet<String>() {{
        this.add("subscribe");
        this.add("psubscribe");
    }};

    // private boolean HaveData;
    //private String Session;
    private final ChannelHandlerContext Writer;
    private final ProcessorRedisImp Processer;
    //private final StringBuilder Buffer;
    private final ArrayList<String> Data = new ArrayList<>();

    private volatile int TableId = 0;

    private volatile AclAuthen Acl = null;

    private volatile long Timestamp;

    private volatile List Request;

    private final BlockingQueue<List> RequestQueue = new ArrayBlockingQueue<>(10000);

    public DefaultDataResultImp(ChannelHandlerContext ctx, ProcessorRedisImp messageProcesser) {
        this.Writer = ctx;
        this.Processer = messageProcesser;
    }

    public synchronized void put(List data) throws InterruptedException {
        boolean empty = RequestQueue.isEmpty();
        RequestQueue.put(data);
        if (empty) {
            process();
        }
    }

    private void process() {
        List data = RequestQueue.peek();
        while (data != null && data.size() > 0) {
            String cmd = (String) data.get(0);
            try {
                if (!this.Processer.process(data, this)) {
                    Writer.close();
                }

                // 此处影响输出统计日志
                ProcessCounter.increase();
            } catch (Throwable t) {
                Writer.writeAndFlush("-ERR " + t.getMessage());
                logger.warnLog("DefaultDataResultImp::process() A fatal error occur: {}", t);
            } finally {
                if (!UnCallbackCommand.contains(cmd)) {
                    // 如果命令不在在UnCallbackCommand列表中，说明该命令会异步执行并调用callback
                    // 否则，当前执行的命令不会触发callback，要继续执行下一条命令
                    break;
                }
            }
        }
    }

    @Override
    public ChannelHandlerContext getSession() {
        return this.Writer;
    }

    @Override
    public synchronized void init(Object session, List request) {
        // TODO Auto-generated method stub
        Data.clear();
        Timestamp = System.currentTimeMillis();
        Request = request;
        //Buffer.setLength(0);
    }

    @Override
    public synchronized void flush() throws IOException {
    }

    @Override
    public void setTableId(int id) {
        TableId = id;
    }

    @Override
    public int getTableId() {
        return TableId;
    }

    @Override
    public void setAcl(AclAuthen acl) {
        this.Acl = acl;
    }

    @Override
    public AclAuthen getAcl() {
        return this.Acl;
    }

    @Override
    public void aclAuth(String cmd, byte[] key) {
        if (this.Acl != null) {
            if (!this.Acl.authentication(cmd, key)) {
                throw DataResult.ACL_FAILED_EXCEPTION;
            }
        }
    }

    @Override
    public synchronized void setOk() throws IOException {
        Writer.writeAndFlush("+OK");
    }

    @Override
    public synchronized void setOk(long l) throws IOException {
        // TODO Auto-generated method stub

        Writer.writeAndFlush(new Long(l));
    }

    @Override
    public synchronized void setErr(int err_code, String msg)
            throws IOException {
        // TODO Auto-generated method stub

        Writer.writeAndFlush("-" + msg);
    }

    @Override
    public synchronized ChannelFuture send(String msg) throws IOException {
        // TODO Auto-generated method stub
        return Writer.writeAndFlush(msg);
    }

    @Override
    public synchronized void sendObject(Object o) {
        // TODO Auto-generated method stub
        Writer.writeAndFlush(o);
    }

    @Override
    public synchronized void callback(Object o) {
        sendObject(o);
        RequestQueue.poll();
        long consuming = System.currentTimeMillis() - Timestamp;
        if (consuming >= ProxyConfig.getSlowOperationThreshold()) {
            try {
                if (Writer != null) {
                    SessionAttribute attribute = CachedSessionAttributes.get(Writer);
                    SlowLogs.add(Request, consuming, Writer, attribute != null ? attribute.getClientName() : null);
                }
            } catch (Throwable t) {
            }
        }
        process();
    }

    @Override
    public String toString() {
        return "Redis Compatible connection to " + Writer.channel().remoteAddress();
    }
}
